/*
 * Copyright 2019 Spotify AB.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.spotify.scio.examples.extra

import com.spotify.scio.bigquery._
import com.spotify.scio.testing._

final class TypedStorageBigQueryTornadoesTest extends PipelineSpec {
  import TypedStorageBigQueryTornadoes.{Result, Row}

  val inData: Seq[Row] = Seq(
    Row(1, Some(true)),
    Row(3, Some(true)),
    Row(4, Some(true)),
    Row(4, Some(true))
  )

  val expected: Seq[Result] = Seq(Result(1, 1), Result(3, 1), Result(4, 2))

  "StorageTypedBigQueryTornadoes" should "work" in {
    JobTest[com.spotify.scio.examples.extra.TypedStorageBigQueryTornadoes.type]
      .args("--output=dataset.table")
      .input(
        BigQueryIO(
          TypedStorageBigQueryTornadoes.Row.table,
          List("tornado", "month"),
          Some("tornado = true")
        ),
        inData
      )
      .output(BigQueryIO[Result]("dataset.table")) { coll =>
        coll should containInAnyOrder(expected)
      }
      .run()
  }
}
